package org.cancer.service

import org.apache.spark.streaming.dstream.DStream
import org.cancer.bean.CancerTypeDeathData_Faye
import org.cancer.common.TService
import org.cancer.dao.CancerTypeDeathDao

class CancerTypeDeathService_Faye extends TService[DStream[CancerTypeDeathData_Faye]]{
  override def dataAnalysis(data: DStream[CancerTypeDeathData_Faye]): Unit = {
    CancerTypeDeathNum(data)
  }

  def CancerTypeDeathNum(data: DStream[CancerTypeDeathData_Faye]): Unit = {
    val deathCounts = data.map(line => {
      (line.cancerType, line.status)
    }).filter(_._2 == "已故") // 只关心已故的数据
      .map(x => (x._1, 1)) // 转换为(cancer_type, 1)

    // 使用 updateStateByKey 维护每个癌症类型的累计死亡人数
    val totalDeathCounts = deathCounts.updateStateByKey[Int]((newCounts: Seq[Int], state: Option[Int]) => {
      val newCount = newCounts.sum
      val previousCount = state.getOrElse(0)
      Some(previousCount + newCount)
    })

    // 输出结果
    totalDeathCounts.foreachRDD { rdd =>
      val sortedRDD = rdd.sortBy(_._2, false)
      sortedRDD.foreachPartition { partition =>
        val dao = new CancerTypeDeathDao()
        partition.foreach { case (cancerType, deathCount) =>
          dao.updateOrInsertDeathCount(cancerType, deathCount)
        }
      }
    }
  }
}
